/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.zookeeper.server.quorum;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.Socket;
import java.nio.ByteBuffer;
import java.util.LinkedList;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;

import org.apache.jute.BinaryInputArchive;
import org.apache.jute.BinaryOutputArchive;
import org.apache.jute.Record;
import org.apache.zookeeper.KeeperException.SessionExpiredException;
import org.apache.zookeeper.ZooDefs.OpCode;
import org.apache.zookeeper.server.ByteBufferInputStream;
import org.apache.zookeeper.server.Request;
import org.apache.zookeeper.server.ZooTrace;
import org.apache.zookeeper.server.quorum.Leader.Proposal;
import org.apache.zookeeper.server.quorum.QuorumPeer.LearnerType;
import org.apache.zookeeper.server.util.SerializeUtils;
import org.apache.zookeeper.server.util.ZxidUtils;
import org.apache.zookeeper.txn.TxnHeader;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * There will be an instance of this class created by the Leader for each
 * learner. All communication with a learner is handled by this
 * class.
 */
public class LearnerHandler extends Thread {
    private static final Logger LOG = LoggerFactory.getLogger(LearnerHandler.class);

    protected final Socket sock;    

    public Socket getSocket() {
        return sock;
    }

    // 对应Leader角色
    final Leader leader;
    // 下一个接收ack的deadline，启动时(数据同步)是一个标准，完成启动后(正常交互)，是另一个标准
    long tickOfLastAck;
    
    /** 当前这个learner的sid */
    protected long sid = 0;
    
    long getSid(){
        return sid;
    }

    // 当前这个learner的version
    protected int version = 0x1;
    
    int getVersion() {
    	return version;
    }
    
    /** 待发送packet的队列 */
    final LinkedBlockingQueue<QuorumPacket> queuedPackets = new LinkedBlockingQueue<>();

    private BinaryInputArchive ia;

    private BinaryOutputArchive oa;

    private BufferedOutputStream bufferedOutput;

    /**
     *  sock已经连接上了
     * @param sock
     * @param leader
     * @throws IOException
     */
    LearnerHandler(Socket sock, Leader leader) throws IOException {
        super("LearnerHandler-" + sock.getRemoteSocketAddress());
        this.sock = sock;
        this.leader = leader;
        //记录在leader的LearnerHandler集合中
        leader.addLearnerHandler(this);
    }
    
    @Override
    public String toString() {
        StringBuilder sb = new StringBuilder();
        sb.append("LearnerHandler ").append(sock);
        sb.append(" tickOfLastAck:").append(tickOfLastAck());
        sb.append(" synced?:").append(synced());
        sb.append(" queuedPacketLength:").append(queuedPackets.size());
        return sb.toString();
    }

    /**
     * 代表一个关闭shutdown的packet来关闭发送packet的线程
     */
    final QuorumPacket proposalOfDeath = new QuorumPacket();
    /**
     * 默认的learner类型（也叫Follower）,也可以设置为OBSERVER
     */
    private LearnerType  learnerType = LearnerType.PARTICIPANT;
    public LearnerType getLearnerType() {
        return learnerType;
    }

    /**
     * 消费 不断消费发送队列，遇到proposalOfDeath就break，遇到Proposal则进行对应的记录
     *
     * @throws InterruptedException
     */
    private void sendPackets() throws InterruptedException {
        long traceMask = ZooTrace.SERVER_PACKET_TRACE_MASK;
        while (true) {
            try {
                QuorumPacket p;
                p = queuedPackets.poll();
                if (p == null) {
                    // 从queuedPackets队列中取出 QuorumPacket
                    bufferedOutput.flush();
                    p = queuedPackets.take();
                }

                // 如果调用了shutDown
                if (p == proposalOfDeath) {
                    break;
                }
                if (p.getType() == Leader.PING) {
                    // 更新当前proposal的时间统计
                    traceMask = ZooTrace.SERVER_PING_TRACE_MASK;
                }
                // 调用序列化方法 写 QuorumPacket
                // oa是jute底层支持序列化的输出流
                // 实际上就是socket的输出流，跟follower之间的连接的输出流
                // 如果有一个follower崩溃了，对于leader而言，在这里时可以感知到的
                oa.writeRecord(p, "packet");
            } catch (IOException e) {
                // leader感知到一个follower宕机,
                // 1. 关闭socket
                // 2. 数据同步线程关闭
                // 3. learnerHandler线程关闭
                if (!sock.isClosed()) {
                    LOG.warn("Unexpected exception at " + this, e);
                    try {
                        sock.close();
                    } catch(IOException ie) {
                        LOG.warn("Error closing socket for handler " + this, ie);
                    }
                }
                break;
                // 如果follower崩溃，影响不是太大，zk集群基本还能运转的，
                // 所有客户端可以重新连接其他节点，读写请求继续，读请求没有影响，写请求没法那么快达到过半follower返回ack过来，监听器还是可以正常的施加。
            }
        }
    }

    static public String packetToString(QuorumPacket p) {
        if (true) {
            return null;
        }
        String type = null;
        String mess = null;
        Record txn = null;
        
        switch (p.getType()) {
        case Leader.ACK:
            type = "ACK";
            break;
        case Leader.COMMIT:
            type = "COMMIT";
            break;
        case Leader.FOLLOWERINFO:
            type = "FOLLOWERINFO";
            break;    
        case Leader.NEWLEADER:
            type = "NEWLEADER";
            break;
        case Leader.PING:
            type = "PING";
            break;
        case Leader.PROPOSAL:
            type = "PROPOSAL";
            TxnHeader hdr = new TxnHeader();
            try {
                txn = SerializeUtils.deserializeTxn(p.getData(), hdr);
                // mess = "transaction: " + txn.toString();
            } catch (IOException e) {
                LOG.warn("Unexpected exception",e);
            }
            break;
        case Leader.REQUEST:
            type = "REQUEST";
            break;
        case Leader.REVALIDATE:
            type = "REVALIDATE";
            ByteArrayInputStream bis = new ByteArrayInputStream(p.getData());
            DataInputStream dis = new DataInputStream(bis);
            try {
                long id = dis.readLong();
                mess = " sessionid = " + id;
            } catch (IOException e) {
                LOG.warn("Unexpected exception", e);
            }

            break;
        case Leader.UPTODATE:
            type = "UPTODATE";
            break;
        default:
            type = "UNKNOWN" + p.getType();
        }
        String entry = null;
        if (type != null) {
            entry = type + " " + Long.toHexString(p.getZxid()) + " " + mess;
        }
        return entry;
    }


    /**
     * 完成了和Learner的启动时的数据同步，同步完成后进行正常的交互.
     *
     * 1. Leader解析Learner信息，计算新的epoch(leader端)
     * 2. 发送Leader状态。(leader端)
     * 3. 数据同步(leader,learner端)
     * 4. 启动Leader和Learner服务器。(learner和leader)
     *
     * 数据同步
     * 1. 根据Learner发送的lastZxid与自己lastZxid，maxCommittedLog,minCommittedLog作比较
     * 2. 来决定是SNAP，DIFF还是TRUNC，后面往往都会跟着PROPOSAL和COMMIT
     * 3. 发完了之后跟着一个NEWLEADER，代表自己是新的LEADER
     * 4. 再跟着一个UPTODATE，代表需要同步的信息发完了
     * 等收到过半LEARNER回复给NEWLEADER的ACK后，即启动完成，可以和LEARNER完成正常的交互，处理请求
     *
     *
     * 读取并解析follower发送过来的注册数据包
     * 如果follower刚跟leader进行连接、注册之后，会进行数据的同步
     * 数据同步完之后，在这里单独启动一个线程，给follower不停的同步数据
     * leader 不停的给follower发送数据过去， 2pc + 过半写机制
     * 不停的 接收follower接收的ack
     */
    @Override
    public void run() {
        try {
            // 1. Leader接收到Learner服务器基本信息后，会解析出该Learner的SID和ZXID
            ia = BinaryInputArchive.getArchive(new BufferedInputStream(sock.getInputStream())); // 对leader而言，也是使用jute进行序列化和反序列化
            bufferedOutput = new BufferedOutputStream(sock.getOutputStream());
            oa = BinaryOutputArchive.getArchive(bufferedOutput);

            QuorumPacket qp = new QuorumPacket();
            ia.readRecord(qp, "packet"); // 会等待读取follower发送过来的注册数据包
            if(qp.getType() != Leader.FOLLOWERINFO && qp.getType() != Leader.OBSERVERINFO){
            	LOG.error("First packet " + qp.toString() + " is not FOLLOWERINFO or OBSERVERINFO!");
                return;
            }

            // 接收learner发送过来的LearnerInfo,包含sid
            byte learnerInfoData[] = qp.getData();
            if (learnerInfoData != null) {
            	if (learnerInfoData.length == 8) {
            		ByteBuffer bbsid = ByteBuffer.wrap(learnerInfoData);
            		this.sid = bbsid.getLong();
            	} else {
            		LearnerInfo li = new LearnerInfo();
            		ByteBufferInputStream.byteBuffer2Record(ByteBuffer.wrap(learnerInfoData), li);
            		this.sid = li.getServerid();
            		this.version = li.getProtocolVersion();
            	}
            } else {
            	this.sid = leader.followerCounter.getAndDecrement();
            }

            LOG.info("Follower sid: " + sid + " : info : " + leader.self.quorumPeers.get(sid));
                        
            if (qp.getType() == Leader.OBSERVERINFO) {
                  learnerType = LearnerType.OBSERVER;
            }

            // 记录当前learner的最新的epoch
            long lastAcceptedEpoch = ZxidUtils.getEpochFromZxid(qp.getZxid());
            
            long peerLastZxid;
            StateSummary ss = null;
            long zxid = qp.getZxid();
            // 6.如果learner的epoch比自己高，更新自己的
            long newEpoch = leader.getEpochToPropose(this.getSid(), lastAcceptedEpoch);

            // leader是旧版本
            if (this.getVersion() < 0x10000) {
                // 然后根据ZXID解析出对应的epoch_of_learner
                long epoch = ZxidUtils.getEpochFromZxid(zxid);
                ss = new StateSummary(epoch, zxid);
                leader.waitForEpochAck(this.getSid(), ss);
            }
            // leader是新版本
            else {
                byte ver[] = new byte[4];
                ByteBuffer.wrap(ver).putInt(0x10000);
                QuorumPacket newEpochPacket = new QuorumPacket(Leader.LEADERINFO, ZxidUtils.makeZxid(newEpoch, 0), ver, null);
                oa.writeRecord(newEpochPacket, "packet");
                bufferedOutput.flush();
                QuorumPacket ackEpochPacket = new QuorumPacket();
                // 9.接收learner的ACKEPOCH
                ia.readRecord(ackEpochPacket, "packet");
                if (ackEpochPacket.getType() != Leader.ACKEPOCH) {
                    LOG.error(ackEpochPacket.toString() + " is not ACKEPOCH");
                    return;
				}
                ByteBuffer bbepoch = ByteBuffer.wrap(ackEpochPacket.getData());
                ss = new StateSummary(bbepoch.getInt(), ackEpochPacket.getZxid());
                leader.waitForEpochAck(this.getSid(), ss);
            }
            peerLastZxid = ss.getLastZxid();
            
            int packetToSend = Leader.SNAP;
            long zxidToSend = 0;
            long leaderLastZxid = 0;
            long updates = peerLastZxid;

            // 如果follower刚跟leader进行连接、注册之后，会进行数据的同步，
            ReentrantReadWriteLock lock = leader.zk.getZKDatabase().getLogLock();
            ReadLock rl = lock.readLock();
            try {
                rl.lock();
                // 内存中记录的最大事务日志的id
                final long maxCommittedLog = leader.zk.getZKDatabase().getmaxCommittedLog();
                // 内存中记录的最小事务日志的id
                final long minCommittedLog = leader.zk.getZKDatabase().getminCommittedLog();

                // 获取提交的Proposal, packet的type都是Leader.PROPOSAL
                LinkedList<Proposal> proposals = leader.zk.getZKDatabase().getCommittedLog();

                if (proposals.size() != 0) {
                    // 如果learner的zxid在leader的[minCommittedLog, maxCommittedLog]范围内
                    if ((maxCommittedLog >= peerLastZxid) && (minCommittedLog <= peerLastZxid)) {

                        long prevProposalZxid = minCommittedLog;

                        boolean firstPacket=true;

                        //默认是DIFF操作
                        packetToSend = Leader.DIFF;
                        zxidToSend = maxCommittedLog;

                        for (Proposal propose: proposals) {
                            // leader提交的proposal已经被learner处理过了，那么就跳过
                            if (propose.packet.getZxid() <= peerLastZxid) {
                                prevProposalZxid = propose.packet.getZxid();
                                continue;
                            } else {
                                // 第一个发送的packet
                                if (firstPacket) {
                                    firstPacket = false;
                                    // 如果learner有一些leader不知道的请求(正常来说应该是prevProposalZxid == peerLastZxid)
                                    if (prevProposalZxid < peerLastZxid) {
                                        // 让learner回滚
                                        packetToSend = Leader.TRUNC;
                                        zxidToSend = prevProposalZxid;
                                        // 让learner回滚
                                        updates = zxidToSend;
                                    }
                                }
                                // 发送PROPOSAL
                                queuePacket(propose.packet);
                                QuorumPacket qcommit = new QuorumPacket(Leader.COMMIT, propose.packet.getZxid(), null, null);
                                // 让刚刚的PROPOSAL进行COMMIT，让learner同步
                                queuePacket(qcommit);
                            }
                        }
                    }
                    // learner的zxid比leader的大，让learner回滚
                    else if (peerLastZxid > maxCommittedLog) {
                        LOG.debug("Sending TRUNC to follower zxidToSend=0x{} updates=0x{}", Long.toHexString(maxCommittedLog), Long.toHexString(updates));

                        packetToSend = Leader.TRUNC;
                        zxidToSend = maxCommittedLog;
                        updates = zxidToSend;
                    } else {
                        LOG.warn("Unhandled proposal scenario");
                    }
                } else if (peerLastZxid == leader.zk.getZKDatabase().getDataTreeLastProcessedZxid()) {
                    LOG.debug("committedLog is empty but leader and follower " + "are in sync, zxid=0x{}", Long.toHexString(peerLastZxid));
                    packetToSend = Leader.DIFF;
                    zxidToSend = peerLastZxid;
                } else {
                    LOG.debug("proposals is empty");
                }               

                LOG.info("Sending " + Leader.getPacketType(packetToSend));
                leaderLastZxid = leader.startForwarding(this, updates);

            } finally {
                rl.unlock();
            }

            // 生成NEWLEADER的packet,发给learner代表自己需要同步的信息发完了
             QuorumPacket newLeaderQP = new QuorumPacket(Leader.NEWLEADER, ZxidUtils.makeZxid(newEpoch, 0), null, null);
             if (getVersion() < 0x10000) {
                oa.writeRecord(newLeaderQP, "packet");
            } else {
                 // 加入发送队列
                queuedPackets.add(newLeaderQP);
            }
            // 发送NEWLEADER消息
            bufferedOutput.flush();
            // 如果是SNAP同步，获取zxid
            if (packetToSend == Leader.SNAP) {
                zxidToSend = leader.zk.getZKDatabase().getDataTreeLastProcessedZxid();
            }
            oa.writeRecord(new QuorumPacket(packetToSend, zxidToSend, null, null), "packet");
            bufferedOutput.flush();

            // 如果发出snap，代表告知learner进行snap方式的数据同步
            if (packetToSend == Leader.SNAP) {
                // SNAP恢复就是把当前的db的序列化内容发送出去
                leader.zk.getZKDatabase().serializeSnapshot(oa);
                //有特定的签名
                oa.writeString("BenWasHere", "signature");
            }
            bufferedOutput.flush();

            // 把数据同步完之后，在这里单独启动一个线程，给follower不停的同步数据
            new Thread() {
                public void run() {
                    Thread.currentThread().setName( "Sender-" + sock.getRemoteSocketAddress());
                    try {
                        // 不断发送packets直到接受到proposalOfDeath
                        sendPackets();
                    } catch (InterruptedException e) {
                        LOG.warn("Unexpected interruption",e);
                    }
                }
            }.start(); // 启动线程，发送消息
            
            qp = new QuorumPacket();
            ia.readRecord(qp, "packet");
            // Learner接收到NEWLEADER 一定会返回ACK
            if(qp.getType() != Leader.ACK){
                LOG.error("Next packet was supposed to be an ACK");
                return;
            }
            // 如果数据同步完之后，会接收follower的ack
            leader.processAck(this.sid, qp.getZxid(), sock.getLocalSocketAddress());

            // 请求阶段的读取超时时间 为 tickTime * syncLimit
            sock.setSoTimeout(leader.self.tickTime * leader.self.syncLimit);

            synchronized(leader.zk){
                while(!leader.zk.isRunning() && !this.isInterrupted()){
                    leader.zk.wait(20);
                }
            }

            // 发送update代表过半的机器回复了NEWLEADER的ACK
            queuedPackets.add(new QuorumPacket(Leader.UPTODATE, -1, null, null));

            // leader 不停的给follower发送数据过去， 2pc + 过半写机制
            // 这里不停的 接收follower接收的ack
            // 正常交互，处理learner的请求等
            while (true) {
                qp = new QuorumPacket();
                ia.readRecord(qp, "packet");

                long traceMask = ZooTrace.SERVER_PACKET_TRACE_MASK;
                if (qp.getType() == Leader.PING) {
                    traceMask = ZooTrace.SERVER_PING_TRACE_MASK;
                }
                if (LOG.isTraceEnabled()) {
                    ZooTrace.logQuorumPacket(LOG, traceMask, 'i', qp);
                }
                tickOfLastAck = leader.self.tick;


                ByteBuffer bb;
                long sessionId;
                int cxid;
                int type;

                switch (qp.getType()) {
                case Leader.ACK:
                    if (this.learnerType == LearnerType.OBSERVER) {
                    }
                    // 核心方法
                    leader.processAck(this.sid, qp.getZxid(), sock.getLocalSocketAddress());
                    break;
                case Leader.PING:
                    ByteArrayInputStream bis = new ByteArrayInputStream(qp.getData());
                    DataInputStream dis = new DataInputStream(bis);
                    while (dis.available() > 0) {
                        long sess = dis.readLong();
                        int to = dis.readInt();
                        leader.zk.touch(sess, to); //会话管理，激活
                    }
                    break;
                case Leader.REVALIDATE:
                    bis = new ByteArrayInputStream(qp.getData());
                    dis = new DataInputStream(bis);
                    long id = dis.readLong();
                    int to = dis.readInt();
                    ByteArrayOutputStream bos = new ByteArrayOutputStream();
                    DataOutputStream dos = new DataOutputStream(bos);
                    dos.writeLong(id);
                    boolean valid = leader.zk.touch(id, to);
                    if (valid) {
                        try {
                            // 设置owner是当前learnerHandler
                            leader.zk.setOwner(id, this);
                        } catch (SessionExpiredException e) {
                            LOG.error("Somehow session " + Long.toHexString(id) + " expired right after being renewed! (impossible)", e);
                        }
                    }
                    dos.writeBoolean(valid);
                    // 返回是否valid
                    qp.setData(bos.toByteArray());
                    queuedPackets.add(qp);
                    break;
                case Leader.REQUEST:                    
                    bb = ByteBuffer.wrap(qp.getData());
                    sessionId = bb.getLong();
                    cxid = bb.getInt();
                    type = bb.getInt();
                    bb = bb.slice();
                    Request si;
                    if(type == OpCode.sync){
                        si = new LearnerSyncRequest(this, sessionId, cxid, type, bb, qp.getAuthinfo());
                    } else {
                        si = new Request(null, sessionId, cxid, type, bb, qp.getAuthinfo());
                    }
                    si.setOwner(this);
                    // 提交请求
                    leader.zk.submitRequest(si);
                    break;
                default:
                }
            }
        } catch (IOException e) {
            // 网络出现故障
            if (sock != null && !sock.isClosed()) {
                LOG.error("Unexpected exception causing shutdown while sock " + "still open", e);
            	try {
            		sock.close();
            	} catch(IOException ie) {
            	}
            }
        } catch (InterruptedException e) {
            LOG.error("Unexpected exception causing shutdown", e);
        } finally {
            LOG.warn("******* GOODBYE "  + (sock != null ? sock.getRemoteSocketAddress() : "<null>") + " ********");
            shutdown();
        }
    }

    /**
     * 关闭当前handler，sock，以及让sendPackets的异步线程最终停止
     */
    public void shutdown() {
        try {
            queuedPackets.put(proposalOfDeath);
        } catch (InterruptedException e) {
            LOG.warn("Ignoring unexpected exception", e);
        }
        try {
            if (sock != null && !sock.isClosed()) {
                sock.close();
            }
        } catch (IOException e) {
            LOG.warn("Ignoring unexpected exception during socket close", e);
        }
        this.interrupt();
        leader.removeLearnerHandler(this);
    }

    public long tickOfLastAck() {
        return tickOfLastAck;
    }

    /**
     * 发送ping命令，本质就是检测leader与learner是否有proposal超时了(超过指定时长没有收到ack)
     */
    public void ping() {
        long id;
        synchronized(leader) {
            id = leader.lastProposed;
        }
        QuorumPacket ping = new QuorumPacket(Leader.PING, id, null, null);
        queuePacket(ping);
    }

    /**
     * 将packet加入异步发送队列
     * run()方法内部 新建一个线程不停的调用 sendPackets();
     * @param p
     */
    void queuePacket(QuorumPacket p) {
        queuedPackets.add(p);
    }

    /**
     * 是否保持同步
     * @return
     */
    public boolean synced() {
        // 线程活着，且当前周期数<deadline
        return isAlive() && tickOfLastAck >= leader.self.tick - leader.self.syncLimit;
    }
}
